﻿/// Copyright © 2013 Microsoft Corporation. 
/// This project is released under the Microsoft Public License (MS-PL)
///
namespace DataMorph.Processing
{
    using System;
    using System.ComponentModel;
    using System.Linq;
    using System.Threading.Tasks;

    using DataMorph.Common;
    using DataMorph.Common.Contracts;
    using DataMorph.Common.Storage;

    using Microsoft.WindowsAzure.Storage.Blob;

    public class WorkScheduler : IWorkScheduler
    {
        private readonly object lockObject = new object();

        private readonly IWorkQueue workQueue;

        private readonly Func<string, string, ICloudStorageAccount> cloudStorageAccountFactory;

        private int generatedMessageCount;

        private bool cancelled;

        public WorkScheduler(IWorkQueue workQueue, Func<string, string, ICloudStorageAccount> cloudStorageAccountFactory)
        {
            this.workQueue = workQueue;
            this.cloudStorageAccountFactory = cloudStorageAccountFactory;
        }

        public event EventHandler<ProgressChangedEventArgs> ProgressChanged;

        public void ScheduleWork(GenerationParameters generationParameters)
        {
            if (generationParameters == null)
            {
                throw new ArgumentNullException("generationParameters", "Generation parameters must not be null or empty.");
            }

            this.cancelled = false;

            var storageAccount = this.cloudStorageAccountFactory(generationParameters.StorageAccountName, generationParameters.StorageAccountKey);
            var request = new ProcessRequest
            {
                StorageAccountName = storageAccount.GetName(),
                DataSetName = generationParameters.Name,
                TimeRequested = DateTimeOffset.UtcNow
            };

            var blobClient = storageAccount.CreateCloudBlobClient();
            var container = blobClient.GetContainerReference(generationParameters.Name);

            if (!container.Exists())
            {
                throw new InvalidOperationException("Specified container doesn't exist.");
            }

            var directories = Enumerable.Range(1, generationParameters.NumberOfDays).Select(dayNumber => container.GetDirectoryReference(dayNumber.ToString()));
            Parallel.ForEach(directories, directory => this.GenerateWorkRequestsForDay(request, this.workQueue, directory, generationParameters));
        }

        public void Cancel()
        {
            this.cancelled = true;
        }

        private void GenerateWorkRequestsForDay(ProcessRequest request, IWorkQueue workQueue, CloudBlobDirectory directory, GenerationParameters generationParameters)
        {
            BlobContinuationToken continuationToken = null;
            do
            {
                if (this.cancelled)
                {
                    break;
                }

                var blobResultSegment = directory.ListBlobsSegmented(continuationToken);

                var processingWorkRequests = blobResultSegment.Results
                                                              .Cast<CloudBlockBlob>()
                                                              .Select(listBlobItem => new ProcessingWorkRequest
                                                                                        {
                                                                                            StorageAccountName = request.StorageAccountName,
                                                                                            DataSetName = request.DataSetName,
                                                                                            File = listBlobItem.Name,
                                                                                            TimeRequested = request.TimeRequested
                                                                                        }).ToList();

                workQueue.AddWorkRequests(processingWorkRequests);
                continuationToken = blobResultSegment.ContinuationToken;

                lock (this.lockObject)
                {
                    this.generatedMessageCount += processingWorkRequests.Count;
                    var progressPercentage = Convert.ToInt32(((double)this.generatedMessageCount / generationParameters.NumberOfFiles) * 100);

                    this.RaiseProgressChangedEvent(progressPercentage, this.generatedMessageCount);
                }
            } 
            while (continuationToken != null);
        }

        private void RaiseProgressChangedEvent(int progress, int messagesGenerated)
        {
            if (this.ProgressChanged != null)
            {
                this.ProgressChanged(this, new ProgressChangedEventArgs(progress, messagesGenerated));
            }
        }
    }
}